package net.pantolomin.nayo;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The reader for data
 */
class DataReader implements Runnable {
	/** The buffer into which we'll read data when it's available */
	private static final ThreadLocal<ByteBuffer> readBuffer = new ThreadLocal<ByteBuffer>();

	/** The logger */
	private final Logger logger;

	/** The connection */
	private final NioConnection connection;

	/**
	 * Create the reader
	 * 
	 * @param connection
	 *          the connection
	 */
	DataReader(NioConnection connection) {
		this.logger = LoggerFactory.getLogger(DataReader.class);
		this.connection = connection;
	}

	/**
	 * @return the buffer to use for reading
	 */
	private ByteBuffer getBuffer() {
		ByteBuffer buffer = readBuffer.get();
		if (buffer == null) {
			int bufferSize = this.connection.getProvider().getConfiguration().getReadBufferSize();
			if (bufferSize < 1024) {
				bufferSize = 1024;
			} else if (bufferSize > 262144) {
				bufferSize = 262144;
			}
			buffer = ByteBuffer.allocate(bufferSize);
			readBuffer.set(buffer);
		}
		return buffer;
	}

	public void run() {
		// Clear out our read buffer so it's ready for new data
		ByteBuffer buffer = getBuffer();
		buffer.clear();

		// Attempt to read off the channel
		int numRead;
		try {
			numRead = this.connection.getSocketChannel().read(buffer);
		} catch (IOException e) {
			if (this.connection.isConnected()) {
				// The remote forcibly closed the connection, cancel
				// the selection key and close the channel.
				this.logger.error("Failed to read a message", e);
				this.connection.getProvider().getSelectionLoop().closeConnectionByRemote(this.connection.getSocketChannel(), false);
			}
			return;
		}

		if (numRead > 0) {
			// Hand the data off to our worker thread
			this.connection.dataRead(Arrays.copyOf(buffer.array(), numRead));

			// Ask to read again
			this.connection.getProvider().getSelectionLoop().askForRead(this.connection.getSelectionKey());
		} else if (numRead == -1) {
			// Remote entity shut the socket down cleanly. Do the same from our
			// end and cancel the channel.
			this.connection.getProvider().getSelectionLoop().closeConnectionByRemote(this.connection.getSocketChannel(), true);
		}
	}
}